package org.spdata.tsdb.dao.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;

import org.bson.Document;
import org.bson.types.Binary;
import org.spdata.tsdb.client.DownSample;
import org.spdata.tsdb.client.SubQuery;
import org.spdata.tsdb.dao.DpsDAO;
import org.spdata.tsdb.dao.MetricDAO;
import org.spdata.tsdb.entity.DpsPack;
import org.spdata.tsdb.entity.Metric;
import org.spdata.tsdb.entity.Summary;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import com.mongodb.BasicDBObject;
import com.mongodb.client.model.Filters;

@Repository(value="dps")
public class DpsDAOImpl extends BaseDAOImpl<DpsPack> implements DpsDAO {

	@Autowired
	MetricDAO metricDAO ;
	
	@Override
	protected DpsPack readEntityFromDocument(Document doc) {		
		
		long id = doc.getLong("_id");
		DpsPack dpsInHour = new DpsPack(id);
		
		Metric m = metricDAO.get(dpsInHour.getMetricId());
		Binary dps = (Binary) doc.get("dps");
		if(dps!=null){
			byte[] bytes = dps.getData();
			dpsInHour.setData(bytes, m.getType());
		}
		
		Document summaryInDB = (Document) doc.get("summary");
		if(summaryInDB!=null){
			Summary summary = new Summary();
			summary.setAvg(summaryInDB.getDouble("avg").floatValue());
			summary.setBegin((float)summaryInDB.getDouble("begin").floatValue());
			summary.setEnd((float)summaryInDB.getDouble("end").floatValue());
			summary.setMax((float)summaryInDB.getDouble("max").floatValue());
			summary.setMin((float)summaryInDB.getDouble("min").floatValue());
			dpsInHour.setSummary(summary);
		}
		return dpsInHour;
	}
	
	private Long findLastHour() {
		BasicDBObject query = new BasicDBObject();
		BasicDBObject sort = new BasicDBObject();
		BasicDBObject projection = new BasicDBObject();
		projection.append("_id", 1);
		sort.append("_id", -1);
		Document document = super.getCollection().find(query).sort(sort).projection(projection).limit(1).first();
		if(document!=null) {
			Long id = document.getLong("_id");
			DpsPack dps = new DpsPack(id);
			return dps.getHour();
		}else 
			return null;
	}
	
	private BasicDBObject getProjection(DownSample sample) {
		BasicDBObject projection = null;
		if(sample!=null){
			if(sample.getUnit().equals(TimeUnit.HOURS) || sample.getUnit().equals(TimeUnit.DAYS)){
				projection = new BasicDBObject("summary", 1);
			}
		}
		if(projection==null) {
			projection = new BasicDBObject("dps", 1);
		}
		return projection;
	}
	
	private Long[] getHourRange (Long startSecond,Long endSecond) {
		Long[] result = new Long[2];
		result[0] = new Long( startSecond / 3600);
		if(endSecond == null) {
			Long lastHour = this.findLastHour();
			if(lastHour!=null) {
				result[1] = lastHour;
			}
		}else {
			result[1] = new Long(endSecond / 3600);
		}
		return result;
	}
	
	public Map<Metric,List<DpsPack>> findDPS(SubQuery query,Long startSecond,Long endSecond,DownSample sample){
		String metricName = query.getMetric();
		Map<String,String>  tags = query.getTags();
		List<Metric> metrics = metricDAO.findMetrics(metricName, tags);
		Map<Metric,List<DpsPack>> result = new HashMap<Metric,List<DpsPack>>();
		if(metrics!=null && metrics.size()>0) {
			for(Metric metric:metrics) {
				List<DpsPack> dpsPack = this.findDPS(metric.getId(), startSecond, endSecond, sample);
				result.put(metric, dpsPack);
			}
		}
		return result;
	}

	public List<DpsPack> findDPS(Integer metricId, Long startSecond, Long endSecond,DownSample sample) {
		List<DpsPack> list = null;		
		BasicDBObject projection = this.getProjection(sample);
		Long[] hourRange = this.getHourRange(startSecond, endSecond);
		if(hourRange[1]==null) {
			return list;
		}
		List<Long> ids = new ArrayList<Long>();
		for(long i = hourRange[0];i<=hourRange[1];i++){
			long id = this.getRowId(i, metricId);
			ids.add(id);
		}		
		list = super.findList(Filters.in("_id", ids),projection);
		
		return list;
	}
	
	private Long getRowId(Long hour,Integer metricIndex){
		Long id = (hour<<32) + metricIndex;
		return id;
	}
}
